package org.solrcn.hadoop.index;


import java.io.IOException;
import java.net.MalformedURLException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.Random;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.impl.CloudSolrServer;
import org.apache.solr.common.SolrInputDocument;

public class SolrMRCluster2 extends Configured implements Tool {

	public static class MapClass extends Mapper<LongWritable, Text, NullWritable, NullWritable> {

		ArrayList<SolrInputDocument> docs = new ArrayList<SolrInputDocument>();
		private static CloudSolrServer solrServer;
		String fileName;
		String zkHost = "sc1:2181";
		String collection = "collection1";
		int shardNums = 4;
		long timeInMillis;
		final long onehour = 6000000;
		
		public void commit(){

		}
		
		@Override
		public void setup(Context context) {
			zkHost = context.getConfiguration().get("zkhost", "sc1:2181");
			collection = context.getConfiguration().get("collection");
			
			//System.out.println("param::collection:"+collection);
			
			shardNums = context.getConfiguration().getInt("shardNums", 1);

			Calendar calendar = Calendar.getInstance();

			int year = context.getConfiguration().getInt("year", 2013);
			int month = context.getConfiguration().getInt("month", 8);
			int day = context.getConfiguration().getInt("day", 1);
			int hour = context.getConfiguration().getInt("hour", 1);

			calendar.set(year, month, day, hour, 0);
			timeInMillis = calendar.getTimeInMillis();

			InputSplit inputSplit = context.getInputSplit();
			fileName = ((FileSplit) inputSplit).getPath().toString();

			try {
				solrServer = new CloudSolrServer(zkHost);
				solrServer.setDefaultCollection(collection);
				
			} catch (MalformedURLException e) {
				e.printStackTrace();
			}

		}

		@Override
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {

			long rows = Long.parseLong(key.toString());
			
			//1  2  3  4  5  6  7  8 

			int shardid = (int) (rows % shardNums) + 1;
			
			//往一台机器里边写
			
			//System.out.println("shardid:"+shardid);
			long boost = timeInMillis + (long) (Math.random() * 6000000);
			SolrInputDocument doc = new SolrInputDocument();
			String[] str = value.toString().split("\t");
			doc.setField("id", java.util.UUID.randomUUID().toString());
			doc.setField("_shard_", "shard" + shardid);
			doc.setField("time_dt", new Date(boost));
			doc.setField("data_" + "sjs" + "_i", (int) (Math.random() * 5000));

			Random rand = new Random();
			Calendar cal = Calendar.getInstance();
			cal.set(1970, 1, 1);
			long start = cal.getTimeInMillis();
			cal.set(2013, 8, 29);
			long end = cal.getTimeInMillis();			
			
			Date d = new Date(start + (long) (rand.nextDouble() * (end - start)));// 时间类型
			cal.setTime(d);
			cal.getTimeInMillis();
			long time_long = cal.getTimeInMillis(); // long
			int time_int = (int) (cal.getTimeInMillis() / 1000); // int

			doc.setField("time1_dt", d);
			doc.setField("time2_l", time_long);
			doc.setField("time3_i", time_int);

			doc.setField("docValuesInt", time_int);
			doc.setField("docValuesLong", time_long);

			doc.setField("docValuesInt_i", time_int);
			doc.setField("docValuesLong_l", time_long);

			if (str.length >= 32) {

				if (StringUtils.isNotBlank(str[0].trim())) {
					doc.setField("data_" + 1 + "_s", str[0]);
				}

				if (StringUtils.isNotBlank(str[3].trim())) {
					doc.setField("data_" + 4 + "_l", Long.parseLong(str[3]));
				}

				if (StringUtils.isNotBlank(str[4].trim())) {
					doc.setField("data_" + 5 + "_l", Long.parseLong(str[4]));
				}

				if (StringUtils.isNotBlank(str[6].trim())) {
					doc.setField("data_" + 7 + "_l", Long.parseLong(str[6]));
				}

				if (StringUtils.isNotBlank(str[31].trim())) {

					doc.setField("data_" + 32 + "_cjk", str[31]);
					doc.setField("bloomString_cjk1", str[31]);

					doc.setField("data_" + 33 + "_s", str[31]);
					doc.setField("bloomString", str[31]);

					doc.setField("bloomLong_l", time_long);
					doc.setField("bloomLong", time_long);

				}

			}
			
			

			docs.add(doc);
			if (docs.size() > 1000) {

				try {

					solrServer.add(docs);

					solrServer.commit();

				} catch (SolrServerException e) {
					e.printStackTrace();
				}

				// docs.clear();

				docs = new ArrayList<SolrInputDocument>();

			}
		}


		@Override
		public void cleanup(Context context) {
			if (docs.size() > 0) {

				try {
					
						//servers[i].add(docs, 15000);
					//solrServer.commit(true,true);
					solrServer.add(docs);
					solrServer.commit();
					
				} catch (SolrServerException e) {
					e.printStackTrace();
				} catch (IOException e) {
					e.printStackTrace();
				}
				//new    
				//docs = new ArrayList<SolrInputDocument>();
				docs.clear();
			}
			
		}
	}

	public static class Reduce extends Reducer<Text, Text, Text, IntWritable> {
		public int max = -1;

		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
			max = -1;
			for (Text val : values) {
				String[] str = val.toString().split("\t", -2);
				if (Integer.parseInt(str[3]) > max)
					max = Integer.parseInt(str[3]);
			}
			context.write(new Text(key), new IntWritable(max));
		}
	}

	public static class AgePartitioner extends Partitioner<Text, Text> {
		@Override
		public int getPartition(Text key, Text value, int numReduceTasks) {
			String[] str = value.toString().split("\t");
			int age = Integer.parseInt(str[1]);
			if (numReduceTasks == 0) {
				return 0;
			}
			if (age <= 20)
				return 0;
			else if (age > 20 && age <= 50)
				return 1 % numReduceTasks;
			else
				return 2 % numReduceTasks;
		}
	}

	@Override
	public void setConf(Configuration conf) {

	}

	@Override
	public Configuration getConf() {
		return new Configuration();
	}

	public int run(String[] args) throws Exception {
		Configuration conf = getConf();
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if (otherArgs.length < 2) {
			System.err.println("Usage: SolrMRCluster <in> <out> [zkHost] [collection] [shardNums] [year] [month] [day] [hour]");

			//hadoop jar HadoopSolr.jar cc.solr.hadoop.index.SolrMRCluster logs2  111  sc1:2181 collection1 4 2012 8 1 1
			//hadoop jar HadoopSolr.jar cc.solr.hadoop.index.SolrMRCluster logs  222  sc1:2181 collection2 4 2012 8 1 2
			
			
			System.exit(2);
		}
		conf.set("zkhost", otherArgs[2]);
		conf.set("collection", otherArgs[3]);
		conf.set("shardNums", otherArgs[4]);
		conf.set("year", otherArgs[5]);
		conf.set("month", otherArgs[6]);
		conf.set("day", otherArgs[7]);
		conf.set("hour", otherArgs[8]);
		
		
		
/*		conf.set("zkhost", "sc1:2181");
		conf.set("collection", "collection1");
		conf.set("shardNums", "4");
		conf.set("year", "2013");
		conf.set("month", "8");
		conf.set("day", "29");
		conf.set("hour", "3");
		*/


		Job job = new Job(conf, "Solr Index");
		job.setJarByClass(SolrMRCluster2.class);//??????????????????????????????????????????????????
		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		
/*		FileInputFormat.addInputPath(job, new Path("logs"));
		FileOutputFormat.setOutputPath(job, new Path("333"));
		*/
		
		job.setMapperClass(MapClass.class);
		job.setNumReduceTasks(0);
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(NullOutputFormat.class);
		System.exit(job.waitForCompletion(true) ? 0 : 1);
		return 0;
	}

	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new SolrMRCluster2(), args);//??????????????????
		System.exit(res);
	}

}
